/**
 * Meituan.com Inc.
 * Copyright (c) 2010-2018 All Rights Reserved.
 */
package com.tuyou.cloud.stream.kafka;

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.core.MessageSource;
import org.springframework.messaging.Message;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;

/**
 * <p></p>
 *
 * @author tuyou@meituan.com
 * @version v0.1 2018-12-23 1:11 PM tuyou Exp $$
 */
@EnableBinding({StreamBinders.STREAM1.class, StreamBinders.STREAM2.class})
public class StreamBinders {

    @Bean
    @InboundChannelAdapter(value = STREAM1.STREAM1, poller = @Poller(fixedDelay = "2000"))
    public MessageSource<LocalDateTime> timerMessageSource() {
        return new MessageSource<LocalDateTime>() {
            @Override
            public Message<LocalDateTime> receive() {
                LocalDateTime now = LocalDateTime.now();
                System.out.println("发送一条消息:" + now);
                return new GenericMessage<>(now);
            }
        };
    }

    @Transformer(inputChannel = STREAM1.STREAM1, outputChannel = STREAM2.STREAM2)
    public Object transform(LocalDateTime date) {
        String format = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").format(date);
        System.out.println("转换一条消息：" + format);
        return format;
    }

    @StreamListener(StreamBinders.STREAM1.STREAM1)
    public void receive(List<Object> payload) {
//        if (Math.random() > 0.5) { //失败后，不能移动偏移量，会重新收到消息，直到消费成功
//            logger.error("ERROR:" + payload);
//            throw new NullPointerException();
//        }
        System.out.println("Received:" + payload);
    }

    public interface STREAM1 {
        String STREAM1 = "STREAM1";

        @Output(STREAM1)
        SubscribableChannel output();
    }

    public interface STREAM2 {
        String STREAM2 = "STREAM2";

        @Output(STREAM2)
        SubscribableChannel input();
    }
}